package com.cy.myrxjavademo

import android.util.Log
import com.cy.myrxjavademo.ImgLoader.Callback
import com.cy.myrxjavademo.bean.User
import io.reactivex.Observable
import io.reactivex.ObservableEmitter
import io.reactivex.ObservableOnSubscribe
import io.reactivex.Observer
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.disposables.Disposable
import io.reactivex.functions.Consumer
import io.reactivex.functions.Function
import io.reactivex.schedulers.Schedulers

/**
 * 参考：https://www.cnblogs.com/xiaxveliang/p/12396110.html
 */
class RxJavaDemo {
    companion object{
        const val TAG:String="RxJavaDemo"
    }
    fun test() {

        val observable = Observable.create(object : ObservableOnSubscribe<String> {
            //e: ObservableEmitter 用来发射事件
            @Throws(Exception::class)
            override fun subscribe(e: ObservableEmitter<String>) {
                Log.i(TAG, "subscribe: " + Thread.currentThread())
                //1、“异步线程” 执行耗时操作
                //subscribe方法内在subscribeOn(Schedulers.io())控制的线程执行，所以下面是同步阻塞的就行
                // 如果本身就是异步的话，就没必要调用.subscribeOn(Schedulers.io()) 了
                ImgLoader.load(true,"http://xxx", object : Callback {
                    override fun onSuccess() {
                        e.onNext("onSuccess")
                        //调用了onComplete()，后续再调用 onNext 就收不到了
                        e.onComplete()
                    }

                    override fun onFailed() {
                        e.onError(Exception())
                    }

                })
                // onComplete()前可以发出多个onNext
                e.onNext("3")

            }
        })
            .subscribeOn(Schedulers.io())
            .map(object :Function<String, User>{
                //用于数据流类型切换，从 String 后续都切换到 User 类型来处理
                override fun apply(t: String): User {
                    return User(t)
                }

            })
            .doOnNext(object :Consumer<User>{
                override fun accept(t: User) {
                    //比如对数据进行修正，存储。不改变数据流
                    //示例场景：你的程序取到的 User 并不应该直接显示，而是需要先与数据库中的数据进行比对和修正后再显示。
                    t.name="拦截name"
                }

            })
            .observeOn(AndroidSchedulers.mainThread())


        callSubscribeConsumer(observable)
        callSubscribeObserver(observable)

    }

    private fun callSubscribeConsumer(observable: Observable<User>) {
        observable.subscribe(object : Consumer<User> {
            //订阅事件的观察者实际需要Observer4个方法处理实现的，而使用consumer可以让我们只需关注处理传递得到的数据，系统会创建空方法实例去补充
            override fun accept(t: User?) {

            }

        })
    }

    private fun callSubscribeObserver(observable: Observable<User>) {

        observable
            // 根据源码，只有调用subscribe方法时，创建Observable时的ObservableOnSubscribe接口的subscribe方法才会被回调
            // subscribe 方法收尾，返回void
            // Observer是完整观察者接口。consumer内部只有一个接口
            .subscribe(object : Observer<User> {
                override fun onSubscribe(d: Disposable) {
                    // 订阅线程  订阅的那一刻在订阅线程中执行
                    Log.i(TAG, "onSubscribe: ${Thread.currentThread()}")
                }

                override fun onNext(value: User) {
                    // “主线程”执行的方法
                    Log.i(TAG, "onNext: $value ${Thread.currentThread()}")
                }

                override fun onError(e: Throwable) {
                    // "主线程"执行的方法
                }

                override fun onComplete() {
                    // "主线程"执行的方法
                    Log.i(TAG, "onComplete: ${Thread.currentThread()}")
                }
            })
    }
}